package com.tmsb.sink;

import com.alibaba.fastjson.JSONObject;
import com.tmsb.pojo.UcenterUser;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.Const;
import com.tmsb.utils.ProvinceCodeTrans;
import com.tmsb.utils.RedisCluster;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;


/**
 * Description: 用ucenter的user来获取日活、省日活、日注册、省日注册这四个实时指标的sink
 * Version:1.0.0
 *
 * @author xiekb
 * @date 2020/6/29
 */
public class DauKeyStateSink extends RichSinkFunction<UcenterUser> {
    private ValueState<Integer> state;
    private JedisCluster jedis;
    private SimpleDateFormat sdf;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        ValueStateDescriptor<Integer> keyedstate =
                new ValueStateDescriptor<>("KeyedStateDeduplication", TypeInformation.of(new TypeHint<Integer>() {}));
        // 状态 TTL 相关配置，过期时间设定为 24 小时
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .cleanupInRocksdbCompactFilter(50000L)
                .build();
        // 开启 TTL
        keyedstate.enableTimeToLive(ttlConfig);
        // 从状态后端恢复状态
        state = getRuntimeContext().getState(keyedstate);
        jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");
    }

    @Override
    public void invoke(UcenterUser value, Context context) throws Exception {
        String lastOnlineTimeFormat = sdf.format(new Date(value.getLast_online_time()*1000));
        int lastOnlineTime = Integer.parseInt(lastOnlineTimeFormat);
        Integer stateValue = state.value();
        if(null == stateValue || !stateValue.equals(lastOnlineTime)){
            //城市日活
            String cityId = value.getCity_id()+"";
            if (cityId.length()== Const.CITYID_LENGTH){
                jedis.zincrby(ConfUtil.getProvincesDauRedisKey()+lastOnlineTimeFormat,1, ProvinceCodeTrans.getProvinceKey(cityId));
            }
            //日活
            jedis.hincrBy(ConfUtil.getDauRedisKey(),lastOnlineTimeFormat,1);

            //日注册
            if(Const.BinlogType.INSERT.toString().equalsIgnoreCase(value.getType())){
                jedis.hincrBy(ConfUtil.getDayRegistRedisKey(),sdf.format(new Date(value.getRegister_time()*1000)),1);
                //城市日注册
                if (cityId.length()== Const.CITYID_LENGTH) {
                    jedis.zincrby(ConfUtil.getDayProvinceRegistRedisKey()+lastOnlineTimeFormat,1, ProvinceCodeTrans.getProvinceKey(cityId));
                }
            }

            state.update(lastOnlineTime);
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        jedis.close();
    }
}
